package com.dd.edu.flashsale.service.listener;

import com.dd.edu.flashsale.service.dao.OrderDao;
import com.dd.edu.flashsale.service.entity.Order;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import static com.dd.edu.flashsale.service.listener.RocketMQOrderListener.MQ_FLASHSALE_ORDER;

/**
 * 将mq中的新订单保存到数据库中
 */
@Component
@RocketMQMessageListener(topic = MQ_FLASHSALE_ORDER,
    consumerGroup = "group_flashsale"
//    consumeMode = ConsumeMode.CONCURRENTLY,
//    messageModel = MessageModel.CLUSTERING,
//    secretKey = "*"
)
public class RocketMQOrderListener implements RocketMQListener<Order> {

    public final static String MQ_FLASHSALE_ORDER = "tp_flashSale";
    public final static String MQ_FLASHSALE_PAY = "tp_flashsale_pay";

    @Autowired
    private OrderDao orderDao;

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Override
    public void onMessage(Order order) {
        // todo 向数据库中插入订单
        System.out.println("向数据库中插入记录：" + order);
        orderDao.save(order);

        // 向mq中发送待支付记录，便于超时后监听关闭订单
        final Message message = new Message(MQ_FLASHSALE_PAY, order.getOrderId().getBytes());
        final org.springframework.messaging.Message springMessage = RocketMQUtil.convertToSpringMessage(message);
        rocketMQTemplate.asyncSend(message.getTopic(), springMessage, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("发送待支付MQ消息成功");
            }

            @Override
            public void onException(Throwable throwable) {
                throwable.printStackTrace();
            }
        }, 1000 * 30, 3);
        // timeout 毫秒
    }
}
